package com.hz.cheetah.hub.zookeeper;

import com.hz.cheetah.hub.StateListener;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.IZkStateListener;
import org.I0Itec.zkclient.ZkClient;
import org.apache.zookeeper.Watcher;

import java.util.List;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;

/**
 * Created by yangchuan on 16/6/20.
 * @since 1.0.0
 */
public class ZookeeperClientImp implements ZookeeperClient {

    private ZkClient zkClient;
    private Watcher.Event.KeeperState state;
    private Set<StateListener> stateListeners = new CopyOnWriteArraySet<StateListener>();

    public ZookeeperClientImp(String zkServers, int sessionTimeout, int connectTimeout) {
        zkClient = new ZkClient(zkServers, sessionTimeout, connectTimeout);
        zkClient.subscribeStateChanges(new IZkStateListener() {
            public void handleStateChanged(Watcher.Event.KeeperState state) throws Exception {
                ZookeeperClientImp.this.state = state;
                if (state == Watcher.Event.KeeperState.Disconnected) {
                    stateChanged(StateListener.DISCONNECTED);
                } else if (state == Watcher.Event.KeeperState.SyncConnected) {
                    stateChanged(StateListener.CONNECTED);
                }
            }

            public void handleNewSession() throws Exception {
                stateChanged(StateListener.RECONNECTED);
            }

            /**
             * Called when a session cannot be re-established. This should be used to implement connection
             * failure handling e.g. retry to connect or pass the error up
             *
             * @param error The error that prevents a session from being established
             * @throws Exception On any error.
             */
            @Override
            public void handleSessionEstablishmentError(Throwable error) throws Exception {

            }
        });

    }


    /**
     * 判断节点是否存在
     *
     * @param path
     * @return
     */
    @Override
    public boolean exists(String path) {
        return zkClient.exists(path);
    }

    /**
     * 创建持久节点
     *
     * @param path
     */
    @Override
    public void createPersistent(String path) {
        zkClient.createPersistent(path);
    }

    /**
     * 创建临时有序节点
     *
     * @param path
     */
    @Override
    public String createEphemeralSequential(String path,Object object) {
        return zkClient.createEphemeralSequential(path,object);
    }

    @Override
    public void createEphemeral(String path,Object object) {
        zkClient.createEphemeral(path,object);
    }


    @Override
    public boolean isConnected() {
        return state == Watcher.Event.KeeperState.SyncConnected;
    }

    protected void stateChanged(int state) {
        for (StateListener sessionListener : getSessionListeners()) {
            sessionListener.stateChanged(state);
        }
    }

    @Override
    public void close() {
        zkClient.close();
    }

    @Override
    public void addStateListener(StateListener listener) {
        stateListeners.add(listener);
    }

    @Override
    public void removeStateListener(StateListener listener) {
        stateListeners.remove(listener);
    }

    public Set<StateListener> getSessionListeners() {
        return stateListeners;
    }

    public List<String> subscribeChildChanges(String path, IZkChildListener listener){
        return zkClient.subscribeChildChanges(path,listener);
    }

    public void unsubscribeChildChanges(String path, IZkChildListener childListener){
        zkClient.unsubscribeChildChanges(path,childListener);
    }

    public List<String> getChildren(String parent){
        return zkClient.getChildren(parent);
    }

}
